import java.net.InetSocketAddress;
import java.nio.ByteOrder;
import java.util.concurrent.Executors;

import net.Protocol;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.buffer.HeapChannelBufferFactory;
import org.jboss.netty.channel.Channel;
import org.jboss.netty.channel.ChannelFactory;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;

import data.ISerializable;

import util.ChannelBufferUtil;

public abstract class Query extends SimpleChannelUpstreamHandler {

	public static final byte TYPE_INT = 0;
	public static final byte TYPE_STRING = 1;
	public static final byte TYPE_OBJECT = 2;
	public static final byte TYPE_BYTE = 8;
	
	public static ChannelFactory factory = new NioClientSocketChannelFactory(
			Executors.newCachedThreadPool(),
			Executors.newCachedThreadPool());
	
	public Query next;
	
	public void start() {
		String host = Config.host;
		int port = 12222;
		ClientBootstrap bootstrap = new ClientBootstrap(factory);
		bootstrap.setPipelineFactory(new RobotPipelineFactory(this));
		bootstrap.setOption("tcpNoDelay", true);
		bootstrap.setOption("bufferFactory", new HeapChannelBufferFactory(
				ByteOrder.LITTLE_ENDIAN));
		
		startTime = System.currentTimeMillis();
		
		bootstrap.connect(new InetSocketAddress(host, port));
		
	}

//	private int _expectLength = Protocol.HEAD_LENGTH;
//	private boolean _readingHead = true;
	public Channel _channel;

	@Override
	public void messageReceived(ChannelHandlerContext ctx, MessageEvent e)
			throws Exception {
		ChannelBuffer data = (ChannelBuffer) e.getMessage();
//		while (_expectLength <= data.readableBytes()) {
//			if (_readingHead) {
//				_expectLength = data.readInt();
//				_readingHead = false;
//			} else {
				System.out.println("query use " + (System.currentTimeMillis()-startTime) + "ms");
				onData(data);
//				_expectLength = Protocol.HEAD_LENGTH;
//				_readingHead = true;
				e.getChannel().close();
				if (listener != null) { 
					listener.onComplete(this);
				}
//				break;
//			}
//		}
	}

	public IQueryListenter listener;

	@Override
	public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) {
		_channel = e.getChannel();
		sendData(e.getChannel());
	}

	abstract public void sendData(Channel channel);

	abstract public void receiveData(ChannelBuffer data);

	public void onData(ChannelBuffer data) throws Exception {
		String msg = ChannelBufferUtil.readString(data);
		if (msg.equals("QueryResult")) {
			data.readInt();
			receiveData(data);
		}
	}

	public int paramIndex;
	public int paramNum;
	public ChannelBuffer buf;

	public void prepare(String service) {
		buf = ChannelBuffers.dynamicBuffer(ByteOrder.LITTLE_ENDIAN, 64);
		buf.writeInt(0);
		ChannelBufferUtil.writeString(buf, "RPC");
		buf.writeInt(0);// qid
		ChannelBufferUtil.writeString(buf, service);
		paramIndex = buf.writerIndex();
		paramNum = 0;
		buf.writeByte(0);
	}

	public void appendParamByte(byte p) {
		buf.writeByte(TYPE_BYTE);
		buf.writeByte(p);
		paramNum++;
	}

	public void appendParamInt(int p) {
		buf.writeByte(TYPE_INT);
		buf.writeInt(p);
		paramNum++;
	}

	public void appendParamString(String p) {
		buf.writeByte(TYPE_STRING);
		ChannelBufferUtil.writeString(buf, p);
		paramNum++;
	}

	public void appendParamObject(ISerializable p, String name) {
		buf.writeByte(TYPE_OBJECT);
		ChannelBufferUtil.writeString(buf, name);
		p.serialize(buf);
		paramNum++;
	}
	
	public long startTime;

	public void send() {
		int len = buf.writerIndex() - Protocol.HEAD_LENGTH;
		buf.setInt(0, len);
		buf.setByte(paramIndex, paramNum);
		_channel.write(buf);
	}
}
